/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.example.quickstart;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.admin.ConsumeStats;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;

/**
 * Consumer，订阅消息
 */
public class Consumer {

	public static void main(String[] args) throws InterruptedException, MQClientException, IOException {
		final AtomicInteger count = new AtomicInteger(1);
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
		/*
		 * Map<String, String> map = new HashMap<String, String>(); consumer.setSubscription(map);
		 */
		String nameServer = "10.230.20.226:8765";
		// consumer.setNamesrvAddr("10.224.66.142:9876;10.224.66.142:8765");
		consumer.setNamesrvAddr(nameServer);
		// consumer.setNamesrvAddr("10.230.20.224:9876");
		// consumer.setInstanceName("instanceNameConsumer");
		/**
		 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
		 * 如果非第一次启动，那么按照上次消费的位置继续消费
		 */
		consumer.subscribe("TopicTest01", "*");
		consumer.setConsumeThreadMin(20); // consumer线程最小值
		consumer.setConsumeThreadMax(64); // consumer线程最大值
		// consumer.setPullBatchSize(32);
		consumer.setConsumeMessageBatchMaxSize(1);
		consumer.setPersistConsumerOffsetInterval(10000000);

		// 从何处开始消费（DefaultMQPushConsumer.consumeFromWhere，
		// 默认为ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET，表示一个新的订阅组第一次启动从队列的最后位置开始消费）；
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		// consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
		// 设置拉取消息后的回调类
		// 在该方法中将参数messageListener赋值给DefaultMQPushConsumer.messageListener和DefaultMQPushConsumerImpl.messageListenerInner变量；
		// 该MessageListener接口有两个实现类MessageListenerConcurrently和 MessageListenerOrderly，具有的业务逻辑在这两个类的consumeMessage方法中实现
		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

				for (MessageExt msg : msgs) {
					int size = msgs.size();
				//	System.out.println("size========" + size);
					// Date d = new Date(msg.getBornTimestamp());
					// System.out.println("============" + msgs.size());

					int value = count.getAndIncrement();
					String body = new String(msg.getBody());
					System.out.println("10.230.20.224:9876===" + body + "消息个数==" + value + "一条消息被消费的次数" + msg.getReconsumeTimes());
					/*if (size >= 3 && count.get() == 2) {
						System.out.println("失败消息" + body + "消息个数==" + value + "一条消息被消费的次数" + msg.getReconsumeTimes());
						try {
							int i = 1 / 0;
						} catch (Exception e) {
							e.printStackTrace();
							return ConsumeConcurrentlyStatus.RECONSUME_LATER;
						}

					}*/
					// System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

				}
				//return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}
		});
		consumer.start();
		System.out.println("Consumer Started.");

		/*
		 * DefaultMQPushConsumer consumer1 = new DefaultMQPushConsumer("ConsumerTEST1_GROUP"); consumer1.setNamesrvAddr("10.230.20.224:9876"); // consumer.setInstanceName("instanceNameConsumer");
		 *//**
			* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
			* 如果非第一次启动，那么按照上次消费的位置继续消费
			*//*
				 * consumer1.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); consumer1.subscribe("TopicTest", "*");
				 * 
				 * consumer1.registerMessageListener(new MessageListenerConcurrently() {
				 * 
				 * @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { for (MessageExt msg : msgs) { System.out.println("222222222" + new String(msg.getBody())); } System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer1.start();
				 * 
				 * System.out.println("Consumer Started.");
				 */
	}
}
